package com.chb.catRecommend;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.Map;

import org.apache.commons.collections.functors.ForClosure;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


public class Step5 {
	public static boolean run(Configuration conf, Map<String, String> paths) throws Exception {
		FileSystem fs = FileSystem.get(conf);
		Job job = Job.getInstance();
		job.setJar("C:\\Users\\12285\\Desktop\\cr.jar");
		job.setJarByClass(Step5.class);
		job.setJobName("Step5");

		job.setMapperClass(Step5Mapper.class);
		job.setReducerClass(Step5Reducer.class);
		job.setSortComparatorClass(NumSort.class);
		job.setGroupingComparatorClass(UserGroup.class);
		
		job.setInputFormatClass(KeyValueTextInputFormat.class); 
		
		job.setMapOutputKeyClass(PairWritable.class);
		job.setMapOutputValueClass(Text.class);
		
		
		Path in = new Path(paths.get("step5Input"));
		FileInputFormat.addInputPath(job, in);
		Path out = new Path(paths.get("step5Output"));
		if (fs.exists(out)) {
			fs.delete(out, true);
		}
		FileOutputFormat.setOutputPath(job, out);
		boolean f = job.waitForCompletion(true);
		return f;
	}

	static class Step5Mapper extends Mapper<Text, Text, PairWritable, Text> {
		@Override
		protected void map(Text key, Text value, Context context)
				throws IOException, InterruptedException {
			String userId = key.toString().split(":")[0];
			String itemId = key.toString().split(":")[1];
			Double num = Double.parseDouble(value.toString());
			
			PairWritable pw = new PairWritable();
			pw.setUserId(userId);
			pw.setNum(num);
			context.write(pw, new Text(itemId+":"+num));
		}
	}
	static class Step5Reducer extends Reducer<PairWritable, Text, Text, Text> {
		 	@Override
		 	protected void reduce(PairWritable key, Iterable<Text> values,
		 			Reducer<PairWritable, Text, Text, Text>.Context context)
		 			throws IOException, InterruptedException {
		 		int i = 0;
		 		StringBuilder sb = new StringBuilder();
		 		for (Text tv : values) {
					if (i > 10) {
						break;
					}
					sb.append(tv.toString()+",");
					i++;
				}
		 		context.write(new Text(key.getUserId()), new Text(sb.toString()));
		 	}
	}
	static class PairWritable implements WritableComparable<PairWritable> {
		private String userId;//按照用户分组
		private Double num;//推荐向量， 降序排序， 获取前10个
		
		@Override
		public void write(DataOutput out) throws IOException {
			out.write(userId.getBytes());
			out.writeDouble(num);
		}

		@Override
		public void readFields(DataInput in) throws IOException {
			this.userId = in.readUTF();
			this.num = in.readDouble();
		}

		@Override
		public int compareTo(PairWritable o) {//这个比较是用来分组比较mapper的输出key是否相同，进行分组
			int r = this.userId.compareTo(o.getUserId());
			if (r == 0) {
				return Double.compare(this.num, o.getNum());
			}
			return 0;
		}

		public String getUserId() {
			return userId;
		}

		public void setUserId(String userId) {
			this.userId = userId;
		}

		public Double getNum() {
			return num;
		}

		public void setNum(Double num) {
			this.num = num;
		}
		 
	
	}
	
	static class NumSort extends WritableComparator {
		public NumSort() {
			super(PairWritable.class, true);
		}
		@Override
		public int compare(WritableComparable a, WritableComparable b) {
			PairWritable p1 = (PairWritable)a;
			PairWritable p2 = (PairWritable)b;
			int r = p1.getUserId().compareTo(p2.getUserId());
			if (r == 0) {//降序排序
				return - Double.compare(p1.getNum(), p2.getNum());
			}
			return r;
		}
	}
	static class UserGroup extends WritableComparator {
		public UserGroup() {
			super(PairWritable.class, true);
		}
		@Override
		public int compare(WritableComparable a, WritableComparable b) {
			PairWritable p1 = (PairWritable)a;
			PairWritable p2 = (PairWritable)b;
			int r = p1.getUserId().compareTo(p2.getUserId());
			return r;
		}
	}
	
	
}
